//  Copyright (c) 2020-present,  INSPUR Co, Ltd.  All rights reserved.
// This source code is licensed under Apache 2.0 License.

// liliupeng@inspur.com

#include "pure_mem/abstract_version_node.h"
#include "util/testharness.h"
#include <chrono>
#include <iostream>
#include "pure_mem/encoded_version_node.h"
#include "pure_mem/uncoded_version_node.h"
#include <pure_mem/memoryblock/memory_arena/memory_arena.h>
#include <string>

namespace rocksdb {

Slice genMVCCKey1(int s) {
  std::string key = "test_" + std::to_string(s);
  uint32_t keySize = key.size() + 1;
  char* ret = new char[keySize];
  memset(ret, '\0', keySize);
  memcpy(ret, key.data(), key.size());
  return {ret, keySize};
}

Slice genInternalKey(Slice& key, SequenceNumber seq, ValueType t, char* buf){
  size_t  key_size = key.size();
  char* p = buf;
  memcpy(p, key.data(), key_size);
  p += key_size;
  uint64_t packed = PackSequenceAndType(seq, t);
  EncodeFixed64(p, packed);
  return {buf, key_size + 8};
}

class VersionNodeExample {
public:
  ArtTree* list_;
  int nodeType_;
public:
  VersionNodeExample(ArtTree* list, int nodeType) : list_(list), nodeType_(nodeType) {}

  ~VersionNodeExample() {
    auto iter = list_->GetIterator();
    iter->SeekToFirst();
    while(iter->Valid()){
      deleteNode(nodeType_, iter->node());
      iter->Next();
    }
  }


  void testNode(int nodeType) {
    std::cout << "test case: generate node!" << std::endl;

    char tmp[1000];
    memset(tmp, 70, 1000);
    Slice value = Slice(tmp, 1000);
    SequenceNumber seq = 100000;
    for(int i = 0; i < 1000; i++){
      Slice key = Slice(genMVCCKey1(seq++));
      VersionNode *node = genNode(nodeType, key, value, seq, kTypeValue);
      bool ok = list_->Insert(reinterpret_cast<const char *>(node));
      assert(ok);
    }
  }

  void testGet(int nodeType) {
    std::cout << "test case: get node's information!" << std::endl;
    char tmp[1000];
    memset(tmp, 70, 1000);
    Slice value = Slice(tmp, 1000);
    SequenceNumber seq = 100000;
    MvccKey mk;
    SequenceNumber s;
    ValueType t;
    Slice cur_value;
    for(int i = 0; i < 1000; i++){
      Slice key = Slice(genMVCCKey1(seq++));
      VersionNode *node = genNode(nodeType, key, value, seq, kTypeValue);
      bool ok = list_->Insert(reinterpret_cast<const char *>(node));
      assert(ok);
      auto iter = list_->GetIterator();

      size_t key_size = key.size();
      size_t internal_key_size = key_size + 8;
      char buf[internal_key_size];
      Slice internal_key = genInternalKey(key, seq, kTypeValue, buf);

      iter->Seek(internal_key);
      assert(iter->Valid());
      auto cur_node = iter->node();
      cur_node->GetInternalKey(internal_key, iter->key());
      cur_node->GetMvccKey(mk, internal_key);
      cur_node->GetValue(internal_key, cur_value);
      cur_node->GetSeqAndType(mk.seqNumAndType_, &s, &t);
//      UnPackSequenceAndType(mk.seqNumAndType_, &s, &t);
      ASSERT_EQ(mk.userKeyWithTime_, key);
      ASSERT_EQ(cur_value, value);
      ASSERT_EQ(s, seq);
      ASSERT_EQ(t, kTypeValue);
    }

    seq = 100000;
    for(int i = 0; i < 1000; i++){
      Slice key = Slice(genMVCCKey1(seq++));
      size_t key_size = key.size();
      size_t internal_key_size = key_size + 8;
      char buf[internal_key_size];
      Slice internal_key = genInternalKey(key, seq, kTypeValue, buf);

      auto iter = list_->GetIterator();
      iter->Seek(internal_key);
      auto cur_node = iter->node();
      cur_node->GetInternalKey(internal_key, iter->key());
      cur_node->GetMvccKey(mk, internal_key);
      cur_node->GetValue(internal_key, cur_value);
      cur_node->GetSeqAndType(mk.seqNumAndType_, &s, &t);
//      UnPackSequenceAndType(mk.seqNumAndType_, &s, &t);
      ASSERT_EQ(mk.userKeyWithTime_, key);
      ASSERT_EQ(cur_value, value);
      ASSERT_EQ(s, seq);
      ASSERT_EQ(t, kTypeValue);
    }
  }


  VersionNode *genNode(int type, const Slice &key, const Slice &value,
                          SequenceNumber seq, ValueType valuetype) {
      switch (type) {
        case 1:{
          auto* cur = new EncodedVersionNode();

          size_t key_size = key.size();
          size_t val_size = value.size();
          size_t interkey_size = key_size + 8;
          size_t encoded_len = VarintLength(interkey_size) + interkey_size +
                         VarintLength(val_size) + val_size;

          char* buf = new char[encoded_len];
          VersionNode::Encode(key, value, seq, valuetype, buf);
          cur->SetNext(nullptr);
          cur->SetPrev(nullptr);
          bool ok = cur->CASSetKey(nullptr, buf);
          assert(ok);
          return cur;
        }
        case 2:
          return new UncodedVersionNode(key, value, seq, valuetype);

        default:
          break;
      }
      return nullptr;
    }

    void deleteNode(int type, VersionNode * node) {
      switch (type) {
        case 1:
          delete[] node->Key();
          break;
        case 2:
          break;
        default:
          break;
      }
    }

};

class NodeTest : public testing::Test, public testing::WithParamInterface<int> {
public:
};

INSTANTIATE_TEST_CASE_P(artTreType, NodeTest,testing::Values(1, 2));

TEST_P(NodeTest, testNode) {
  InternalKeyComparator cmp(BytewiseComparator());
  MemArena::KeyComparator mkey_cmp(cmp);
  ArtTree list(mkey_cmp, ART_ROWEX);
  int nodeType = GetParam();
  VersionNodeExample vn(&list, nodeType);
  vn.testNode(nodeType);
}

TEST_P(NodeTest, testGet) {
  InternalKeyComparator cmp(BytewiseComparator());
  MemArena::KeyComparator mkey_cmp(cmp);
  ArtTree list(mkey_cmp, ART_ROWEX);
  int nodeType = GetParam();
  VersionNodeExample vn(&list, nodeType);
  vn.testGet(nodeType);
}

} // namespace rocksdb

int main(int argc, char **argv) {
  ::testing::InitGoogleTest(&argc, argv);
  return RUN_ALL_TESTS();
  system("pause");
}
